package cn.doitedu.rtmk.demo5;

import cn.doitedu.rtmk.common.EventBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * 规则id: rule-1：
 * 静态画像条件： age:[20,39] , gender = male
 * 实时动态画像标签条件: w行为次数 > 3
 * 触发条件: u行为
 */

/**
 * 规则id: rule-2：
 * 静态画像条件： age:[25,30] , active_level>2
 * 实时动态画像标签条件: g行为次数 >=4
 * 触发条件: p行为
 */

public class RuleModel_1_Calculator implements RuleCalculator {
    Table userProfileTable;
    JSONObject jsonObject;
    ValueState<Integer> rule1State;

    JSONObject paramJsonObject;
    String ruleId;


    @Override
    public void init(RuntimeContext runtimeContext, String ruleParamJsonStr) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "doitedu:2181");

        Connection hbaseConnection = ConnectionFactory.createConnection(conf);
        userProfileTable = hbaseConnection.getTable(TableName.valueOf("user_profile"));

        jsonObject = new JSONObject();


        // 申请一个keyedState，来记录每个用户的规则1的w行为次数
        rule1State = runtimeContext.getState(new ValueStateDescriptor<Integer>("rule-1-state", Integer.class));

        // 将规则参数，解析成fastjson内部的通用的对象
        paramJsonObject = JSON.parseObject(ruleParamJsonStr);
        ruleId = paramJsonObject.getString("rule_id");


    }

    @Override
    public void calc(EventBean eventBean, Collector<String> collector) throws IOException {

        String eventId = eventBean.getEventId();

        // 0. 判断事件是否是动态画像的目标事件，如果是，则做动态画像统计
        JSONObject realtimeProfileObject = paramJsonObject.getJSONObject("realtime_profile");
        String realtimeProfileEventId = realtimeProfileObject.getString("eventId");
        if (eventId.equals(realtimeProfileEventId)) {
            rule1State.update( ( rule1State.value() == null ? 0 : rule1State.value()) + 1);
        }

        // 1.判断事件是否是触发条件
        String fireActionEventId = paramJsonObject.getString("fire_action");
        if (eventId.equals(fireActionEventId)) {
            // 2.判断该用户的 “动态画像条件”是否已满足
            Integer realtimeEventCnt = realtimeProfileObject.getInteger("eventCnt");
            if (rule1State.value() != null && rule1State.value() >= realtimeEventCnt) {

                // 3.1 从参数中取静态画像条件的参数，拼装成 hbase的 get 条件
                Get getParam = new Get(Bytes.toBytes(eventBean.getUid()));
                JSONArray staticProfileJsonArr = paramJsonObject.getJSONArray("static_profile");
                for (int i = 0; i < staticProfileJsonArr.size(); i++) {
                    JSONObject obj_i = staticProfileJsonArr.getJSONObject(i);
                    String tagName = obj_i.getString("tagName");
                    getParam.addColumn(Bytes.toBytes("f"), Bytes.toBytes(tagName));

                }

                // 3.2 根据拼装好的get条件，去hbase查询用户的标签值
                Result result = userProfileTable.get(getParam);

                // 3.3 遍历静态画像的每一个条件，去比较： 参数要求的标签值 和  hbase中查询出来的值  是否匹配
                for (int i = 0; i < staticProfileJsonArr.size(); i++) {
                    // 从参数中取出要求的  标签名、运算符、标签值
                    JSONObject obj_i = staticProfileJsonArr.getJSONObject(i);

                    // 是参数中要求的标签名
                    String tagName = obj_i.getString("tagName");
                    // 是参数中该标签要求的值
                    JSONArray paramTagValue = obj_i.getJSONArray("tagValue");
                    // 是参数中要求的比较类型
                    String compareType = obj_i.getString("compareType");

                    // 从hbase返回的结果中取出对应的标签值
                    byte[] tagValueBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes(tagName));
                    String tagValueStr = Bytes.toString(tagValueBytes);

                    // 比较当前遍历的参数要求tagValue和hbase查出来的tagValue是否匹配
                    boolean b = valueCompare(paramTagValue, compareType, tagValueStr);

                    if (!b) return;
                }

                // 3.4 如果上面的比较全部通过，则说明静态画像条件全部满足，则输出命中消息
                jsonObject.put("uid", eventBean.getUid());
                jsonObject.put("timestamp", eventBean.getTimestamp());
                jsonObject.put("rule_id", ruleId);

                collector.collect(jsonObject.toJSONString());

            }

        }


    }


    private boolean valueCompare(JSONArray paramTagValue, String compareType, String tagValueStr) {

        boolean res = false;
        String minStr;
        String maxStr;
        double min;
        double max;
        double realValue;
        switch (compareType) {
            case "between":
                minStr = paramTagValue.getString(0);
                maxStr = paramTagValue.getString(1);

                min = Double.parseDouble(minStr);
                max = Double.parseDouble(maxStr);
                realValue = Double.parseDouble(tagValueStr);

                res = realValue >= min && realValue <= max;
                break;
            case ">":
                minStr = paramTagValue.getString(0);
                min = Double.parseDouble(minStr);
                realValue = Double.parseDouble(tagValueStr);

                res = realValue > min;
                break;

            case "<":
                maxStr = paramTagValue.getString(0);
                max = Double.parseDouble(maxStr);
                realValue = Double.parseDouble(tagValueStr);

                res = realValue < max;
                break;

            case "=":
                String paramValueStr = paramTagValue.getString(0);
                res = tagValueStr.equals(paramValueStr);
                break;
            default:
        }

        return res;
    }


}
